package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo5Shebao {
  def main(args: Array[String]): Unit = {

    /**
      * 1、统计人的工作履历
      *
      * 统计结果
      * 91330000733796106P,杭州海康威视数字技术股份有限公司,2020-02-01 00:00:00,2020-05-01 00:00:00
      * 91330000733796106P,阿里云计算有限公司,2020-06-01 00:00:00,2020-09-01 00:00:00
      * 91330000733796106P,杭州海康威视数字技术股份有限公司,2020-10-01 00:00:00,2021-03-01 00:00:00
      */

    val spark: SparkSession = SparkSession
      .builder()
      .appName("shebao")
      .master("local")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._


    //1、读取数据
    val dataDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING, burk STRING, sdate STRING")
      .load("data/data.txt")

    dataDF
      //1、获取这个人上一个月所在的公司
      .withColumn("lastBurl", lag($"burk", 1) over Window.partitionBy($"id").orderBy("sdate"))
      //2、如果当前月和上一个月是同一个公司在后面打上标记0， 如果不是一个公司打上标记1
      .withColumn("flag", when($"burk" === $"lastBurl", 0).otherwise(1))
      //3、对后面的标记累加求和
      .withColumn("clazz", sum($"flag") over Window.partitionBy($"id").orderBy($"sdate"))
      //4、按照id，公式，clazz分组
      .groupBy($"id", $"burk", $"clazz")
      //获取入职时间和离职时间
      .agg(min($"sdate") as "start_date", max($"sdate") as "end_date")
      .show()
  }
}
